package club.monkeywood.ad.dmp.report

import java.util.Properties

import com.typesafe.config.ConfigFactory
import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}


object ProCityRptV2 {


    def main(args: Array[String]): Unit = {

        // 0 校验参数个数
        if (args.length != 1) {
            println(
                """
                  |club.monkeywood.ad.dmp.ProCityRptV2
                  |参数：
                  | logInputPath
                """.stripMargin)
            sys.exit()
        }

        // 1 接受程序参数
        val Array(logInputPath) = args

        // 2 创建sparkconf->SparkSession
        val sparkConf = new SparkConf()
        sparkConf.setAppName(s"${this.getClass.getSimpleName}")
        sparkConf.setMaster("local[*]")
        //使用KryoSerializer更快
        // RDD 序列化到磁盘 worker与worker之间的数据传输
        sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

        val ss = SparkSession.builder()
          .appName(s"${this.getClass.getSimpleName}")
          .master("local[*]")
          .config(sparkConf)
          .getOrCreate()

        // 读取数据 --》parquet文件
        val df: DataFrame = ss.read.parquet(logInputPath)

        // 将dataframe注册成一张临时表
        df.createTempView("log")

        // 按照省市进行分组聚合---》统计分组后的各省市的日志记录条数
        val result: DataFrame = ss.sql("select provincename, cityname, count(*) ct from log group by provincename, cityname")

        // 自动从资源文件中加载配置文件
        // 查找顺序：application.conf -> application.json --> application.properties
        val load = ConfigFactory.load()
	      val url = load.getString("jdbc.url")
	      val tableName = load.getString("jdbc.tableName")
        val props = new Properties()
        props.setProperty("user", load.getString("jdbc.user"))
        props.setProperty("password", load.getString("jdbc.password"))

        // 将结果写入到mysql的 rpt_pc_count 表中
        result.write
	        //插入新数据
	        //默认创建新表
	        .mode(SaveMode.Append)
	        .jdbc(url, tableName, props)

        ss.stop()
    }
}
